package operator.condition;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Predicate;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

/**
 * 一旦条件成立，立即终止
 */
class TakeUntilOperator {

    public static void main(String[] args) {
        takeUntil1();
//        takeUntil2();
    }

    /**
     * 发射Observable发射的数据，直到一个指定的条件不成立(包含临界值)
     */
    static void takeUntil1() {
        Observable.just(1, 2, 3)
                .takeUntil(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        Printer.print("test integer = " + integer);
                        return integer == 2;
                    }
                })
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }

    /**
     * 使用一个标志Observable是否发射数据来判断，当标志Observable没有发射数据时，正常发射数据
     * 而一旦标志Observable发射了一项数据或者发射了一个终止通知则后面的数据都会被丢弃
     */
    static void takeUntil2() {
        Observable.interval(1, TimeUnit.SECONDS)
                .takeUntil(Observable.timer(3, TimeUnit.SECONDS)) //延迟3s
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
